package com.tairanchina.md.elife.rabbit.client.consumer.impl;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.tairanchina.md.elife.rabbit.client.consumer.RabbitConsumerClient;
import com.tairanchina.md.elife.rabbit.task.RabbitMQTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.concurrent.TimeoutException;

/**
 * Created by lxliuxuan on 2016/12/12.
 * Updated by 殷洁 on 2016年12月26日17:34:10
 */
public class TXConsumerRabbitClient extends RabbitConsumerClient
{
    private Logger logger = LoggerFactory.getLogger(TXConsumerRabbitClient.class);

    //获取连接工厂
    public ConnectionFactory getConnectionFactory()
    {

        if (factory == null)
        {
            initConnectionFactory();
        }
        return factory;
    }

    //获取连接，连接保持长连接，不同的用户请求提供不同的chnnel
    public Connection getConnection()
    {

        if (connection == null || !connection.isOpen())
        {
            initConnection();
        }
        return connection;
    }

    public void receiveMessage(String queueName, int num, RabbitMQTask rabbitMQTask, Object... objs) throws Exception
    {
        Channel channel = null;
        try
        {
            //判断连接是否可用，如果不可用，就创建新的连接
            channel = getChannel();

            //设置每个消费者每次接收的消息数
            channel.basicQos(num);
            QueueingConsumer consumer = new QueueingConsumer(channel);

            //false 表示手动控制应答
            channel.basicConsume(queueName, false, consumer);
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();//得到交付

            //得到交付的消息
            Object obj = handlerDecode(delivery.getBody());

            //处理业务逻辑
            rabbitMQTask.consumerMsg(obj, objs);

            //确认消息已经收到
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
        catch (Exception e)
        {
            e.printStackTrace();
            logger.error("消息接收失败。e:{}", e);
            throw new Exception(e);
        }
        finally
        {
            try
            {
                if (channel != null)
                {
                    channel.close();
                }
            }
            catch (TimeoutException | IOException e)
            {
                e.printStackTrace();
                logger.error("消费者端消息通道关闭失败。e:{}", e);
            }
        }
    }

    /**
     * 消息解码算法
     *
     * @param bytes
     * @return
     */
    protected Object handlerDecode(byte[] bytes) throws Exception
    {
        ByteArrayInputStream bin = null;
        ObjectInputStream oin = null;
        Object object = null;
        try
        {
            bin = new ByteArrayInputStream(bytes);
            oin = new ObjectInputStream(bin);
            object = oin.readObject();
        }
        catch (Exception e)
        {
            e.printStackTrace();
            logger.error("消息解码失败。e:{}---data:{}", e, bytes);
            throw new Exception(e);
        }
        finally
        {
            try
            {
                if (bin != null)
                {
                    bin.close();
                }
            }
            catch (IOException e1)
            {
                e1.printStackTrace();
                logger.error("数据输入流1关闭失败。e:{}", e1);
            }

            try
            {
                if (oin != null)
                {
                    oin.close();
                }
            }
            catch (IOException e2)
            {
                e2.printStackTrace();
                logger.error("数据输入流2关闭失败。e:{}", e2);
            }
        }
        return object;
    }

    public void receiveMessage(String queueName, RabbitMQTask rabbitMQTask, Object... objs) throws Exception
    {
        receiveMessage(queueName, 1, rabbitMQTask, objs);
    }

    public Channel getChannel()
    {
        Channel channel = null;
        try
        {
            channel = getConnection().createChannel();
        }
        catch (IOException e)
        {
            e.printStackTrace();
            logger.error("获取连接通道失败。e:{}", e);
        }
        return channel;
    }

    //初始化RabbitMQ
    public void init()
    {
        initConnectionFactory();
        initConnection();
    }

    public void initConnection()
    {
        try
        {
            connection = getConnectionFactory().newConnection();
        }
        catch (Exception e)
        {
            e.printStackTrace();
            logger.error("RabbitMQ 连接初始化失败 e:{}", e);
        }
    }

    public void initConnectionFactory()
    {
        factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setUsername(username);
        factory.setPassword(password);
        factory.setVirtualHost(virtualhost);
        factory.setPort(port);
        //设置消费者端自动重连
        factory.setAutomaticRecoveryEnabled(true);
    }
}
